package com.shujia.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SparkSession}

object Demo05JDLogAnalysis {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .appName("Demo05JDLogAnalysis")
      .master("local")
      .config("spark.sql.shuffle.partitions", "2")
      .getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._

    val jdLogDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", " ")
      .schema("user_id String,shop String")
      .load("Spark/data/JDLogs/jd.log")

    // 1 每个店铺的UV（访客数），输出店铺名称、访客数
    jdLogDF
      .groupBy($"shop")
      .agg(countDistinct($"user_id") as "cnt")
      .show()

    // 2 每个店铺访问次数top3的访客信息，输出店铺名称、访客id、访问次数
    jdLogDF
      .groupBy($"shop", $"user_id")
      .agg(count("*") as "cnt")
      .withColumn("rn", row_number() over Window.partitionBy($"shop").orderBy($"cnt".desc))
      .where($"rn" <= 3)
      .show()

    val userVisitDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", " ")
      .schema("userID String, visitDate String, visitCount String")
      .load("Spark/data/JDLogs/user_visit.log")


    userVisitDF.cache()
    userVisitDF.createOrReplaceTempView("user_visit")

    // 统计出每个用户每个月的访问次数以及累计访问次数

    userVisitDF
      .withColumn("month", from_unixtime(unix_timestamp($"visitDate", "yyyy/MM/dd"), "yyyyMM"))
      .groupBy($"userID", $"month")
      .agg(sum($"visitCount") as "month_cnt")
      .withColumn("sum_cnt", sum($"month_cnt") over Window.partitionBy($"userID").orderBy($"month"))
      .show()

    spark
      .sql(
        """
          |SELECT  tt1.userID
          |        ,tt1.month
          |        ,tt1.month_cnt
          |        ,sum(tt1.month_cnt) over(partition by tt1.userId order by tt1.month) as sum_cnt
          |from (
          |    SELECT  t1.userID
          |            ,t1.month
          |            ,sum(t1.visitCount) as month_cnt
          |    from (
          |        select  userID
          |                ,visitDate
          |                ,visitCount
          |                ,from_unixtime(unix_timestamp(visitDate, "yyyy/MM/dd"), "yyyyMM") as month
          |        from user_visit
          |    ) t1 group by t1.userID,t1.month
          |) tt1
          |""".stripMargin)
      .show()
  }

}
